package model

import org.apache.spark.ml.PipelineModel
import org.apache.spark.ml.feature.VectorAssembler
import org.apache.spark.sql.{DataFrame, SQLContext, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}

object LinearRegression_online {

  def lrOnline(data_path:String,modelpath:String): Unit = {
    //设置环境
//    val conf=new SparkConf()
//       .setAppName("LinearRegression")
//       .setMaster("local")
//    val sc=new SparkContext(conf)
//    val sqc=new SQLContext(sc)

    val spark: SparkSession = SparkSession
      .builder()
      .appName("LinearRegression")
      .master("local")
      .getOrCreate()
    val sc = spark.sparkContext
    val sqc = spark.sqlContext

    //准备训练集合
    val raw_data=sc.textFile(data_path)
    val map_data=raw_data.map{x=>
      val split_list=x.split(",")
      (split_list(0).toDouble,split_list(1).toDouble,split_list(2).toDouble,split_list(3).toDouble,split_list(4).toDouble,split_list(5).toDouble)
    }

    val df=sqc.createDataFrame(map_data)
    val data = df.toDF("label_y", "feature_x1", "feature_x2", "feature_x3", "feature_x4", "feature_x5")
    val colArray = Array("feature_x1", "feature_x2", "feature_x3", "feature_x4", "feature_x5")
    val assembler = new VectorAssembler().setInputCols(colArray).setOutputCol("features")
    val test: DataFrame = assembler.transform(data)

    // 模型读取
    val lrModel = PipelineModel.load(modelpath)
    println("==============读取模型成功==============")

    //执行预测
    val predictions: DataFrame = lrModel.transform(test)
    val predict_result: DataFrame =predictions.selectExpr("features","label_y", "round(prediction,1) as prediction")

    println("==============输出预测结果==============")
    predict_result.foreach(println(_))

    //结束绘画
    sc.stop()
  }
  def main(args: Array[String]): Unit = {
    //训练数据路径
    //val data_path = "data\\test_data.txt"
    val data_path = "hdfs:///test/data/test_data.txt"
    //读取模型保存路径
    //val saveModelpath = "model\\spark-lr-model"
    val saveModelpath = "hdfs:///test/model/lrModel"
    //LinearRegression
    lrOnline(data_path,saveModelpath)
  }
}
